package com.example.mq.mqServer;

import com.example.mq.common.*;
import com.example.mq.common.basic.*;
import com.example.mq.mqServer.core.message.BasicProperties;
import com.example.mq.mqServer.exception.MqException;
import com.example.mq.mqServer.util.SerializableUtils;
import lombok.extern.slf4j.Slf4j;

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @BelongsProject: 0.mq
 * @BelongsPackage: com.example.mq.mqServer
 * @Author : 游在海里的鱼
 * @CreateTime : 2024-01-03  22:37
 * @Version : 1.0
 * @ClassName : 中间人服务器
 * @Description : 中间人服务器。作为mq服务器和客户端的中间层
 */
@Slf4j
public class BrokerServer {
    //套接字
    private ServerSocket  serverSocket = null;
    //虚拟机-default
    private VirtualHost virtualHost = new VirtualHost("default");
    //会话，用来建立与客户端的连接，逻辑上的建立
    private ConcurrentHashMap<String, Socket> session = new ConcurrentHashMap<String,Socket>();
    //线程池
    private ExecutorService threadPool = null;
    //运行状态，brokerServer服务器的
    private volatile boolean running = true;

    /** ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 构造方法
     * @Describe: TODO
     * @modifier: [public] [构造方法]
    **/
    public BrokerServer(int port){
        try {
            //1.新建brokerServer的套接字
            serverSocket = new ServerSocket(port);
        } catch (IOException e) {
            log.error("[BrokerServer] 创建一个[端口:"+port+"]的[ServerSocket对象]出现异常");
            e.printStackTrace();
        }
    }

    /** ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 启动服务器
     * @Describe: TODO
     * @modifier: [public] [实例方法]
    **/
    public void start(){
        log.info("[start] brokerServer已启动");
        //1.新建线程池
        threadPool = Executors.newCachedThreadPool();
        while (running) {
            try {
                //2.获取客户端
                Socket clientSocket = serverSocket.accept();

                //3.分配线程来处理客户端请求
                threadPool.submit(() -> {
                    processConnection(clientSocket);
                });
            } catch (SocketException e) {

            } catch (IOException e) {
                log.error("[start] 与客户端建立链接失败");
                e.printStackTrace();
            }
        }
    }

    /** ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 关闭服务器
     * @Describe: TODO
     * @modifier: [public] [实例方法]
    **/
    public void close(){
        //1.结束brokerServer
        running = false;
        //2.结束线程池中所有的任务
        threadPool.shutdown();
        try {
        //3.关闭brokerServer的socket
            serverSocket.close();
        } catch (IOException e) {
            log.error("[close] 关闭[serverSocket]失败");
            e.printStackTrace();
        }
        log.info("[close] brokerServer已关闭");
    }

    /* ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 与客户端链接后,处理客户端的请求
     * @Describe: TODO
     * @modifier: [private] [实例方法]
    **/
    private void processConnection(Socket clientSocket) {
        //1.获取客户端的输入输出流
        try(InputStream inputStream = clientSocket.getInputStream();
            OutputStream outputStream = clientSocket.getOutputStream()) {
        //2.给客户端的输入输出流套上Data流
            try (DataInputStream dataInputStream = new DataInputStream(inputStream);
                DataOutputStream dataOutputStream = new DataOutputStream(outputStream)){
                while (true){
        //3.获取请求,从客户端中
                    Request request = readRequest(dataInputStream,clientSocket);
                    if(request == null){
                        break;
                    }
        //4.获取响应,处理请求后
                    Response response = process(request,clientSocket);
        //5.发送响应,往客户端里
                    writeResponse(response,dataOutputStream);
                }
            }
        //客户端请求处理完毕,抛出EOF异常,此异常并非真异常
        } catch (EOFException | SocketException e) {
            log.info("[processConnection] [IP为:"+clientSocket.getInetAddress().toString()+"] " +
                    "[端口为:"+clientSocket.getPort()+"]的客户端事务处理完毕,关闭连接");
        }catch (Exception e) {
            log.error("[processConnection] [IP为:"+clientSocket.getInetAddress().toString()+"] " +
                    "[端口为:"+clientSocket.getPort()+"]的客户端出现异常");
            e.printStackTrace();
        }finally {
            try {
        //6.关闭客户端
                clientSocket.close();
        //7.清除所有channel，当客户端关闭后
                clearClosedSession(clientSocket);
            } catch (IOException e) {
                log.error("[processConnection] [IP为:"+clientSocket.getInetAddress().toString()+"] " +
                        "[端口为:"+clientSocket.getPort()+"]的客户端关闭失败");
                e.printStackTrace();
            }
        }

    }

    /* ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 获取request,从客户端的输入流中
     * @Describe: TODO
     * @modifier: [private] [实例方法]
    **/
    private Request readRequest(DataInputStream dataInputStream ,Socket clientSocket) {
        try {
            //1.新建request对象
            Request request = new Request();
            //2.获取方法类型,从客户端输入流中
            request.setType(dataInputStream.readInt());
            //3.获取payload长度,从客户端输入流中
            request.setLength(dataInputStream.readInt());
            //4.获取payload,从客户端输入流中
            byte[] payload = new byte[request.getLength()];
            int n = dataInputStream.read(payload);
            if(n != request.getLength()){
                new MqException("[readRequest] [IP为:"+clientSocket.getInetAddress().toString()+"]" +
                        " [端口为:"+clientSocket.getPort()+"]的客户端,发出的[请求格式]错误").printStackTrace();
                return null;
            }
            request.setPayload(payload);
            //5.返回request
            return request;
        }catch(EOFException e){

        } catch (IOException e) {
            log.error("[readRequest] 解析请求时,客户端输入流出现异常");
            e.printStackTrace();
        }
        return null;
    }

    /* ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 清除已关闭频道的会话
     * @Describe: TODO
     * @modifier: [private] [实例方法]
    **/
    private void clearClosedSession(Socket clientSocket) {
        //1.记录要删除的channel
        List<String> toDeleteChannel = new ArrayList<>();
        //2. 遍历记录要删除的channel
        for(Map.Entry<String, Socket> entry: session.entrySet()){
            if(entry.getValue() == clientSocket){
                //session.remove(entry.getKey()); 不能一边遍历一边删除同个对象，会导致遍历器失效
                toDeleteChannel.add(entry.getKey());
            }
        }
        //3.删除session，根据记录下的channel
        for(String key:toDeleteChannel){
            session.remove(key);
        }
    }

    /* ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 返回response,往客户端输出流中
     * @Describe: TODO
     * @modifier: [private] [实例方法]
    **/
    private void writeResponse(Response response, DataOutputStream dataOutputStream){
        try {
            //1.写入response
            dataOutputStream.writeInt(response.getType());
            dataOutputStream.writeInt(response.getLength());
            dataOutputStream.write(response.getPayload());
            //2.冲刷输出流
            dataOutputStream.flush();
        } catch (IOException e) {
            log.error("[writeResponse] 写入响应时,客户端输出流出现异常");
            e.printStackTrace();
        }
    }

    /* ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 获取response,通过处理request得到
     * @Describe: TODO
     * @modifier: [private] [实例方法]
    **/
    private Response process(Request request, Socket clientSocket) throws IOException, ClassNotFoundException {
        //1.获取basicArguments，从request中
        BasicArguments basicArguments = (BasicArguments) SerializableUtils.fromBytes(request.getPayload());
        //2.定义方法执行后的返回结果，virtualHost执行的方法
        boolean ok = true;
        //3.执行virtualHost的方法，根据request里的type
        switch(request.getType()){
            //创建channel
            case 0x1:
                session.put(basicArguments.getChannelId(),clientSocket);
                log.info("[process] [IP为"+clientSocket.getInetAddress().toString()+"的客户端]创建了" +
                        "[channel:"+basicArguments.getChannelId()+"]");
                break;
            //删除channel
            case 0x2:
                session.remove(basicArguments.getChannelId());
                log.info("[process] [IP为"+clientSocket.getInetAddress().toString()+"的客户端]删除了" +
                        "[channel:"+basicArguments.getChannelId()+"]");
                break;
            //新建交换机
            case 0x3:
                log.info("[process] [IP为"+clientSocket.getInetAddress().toString()+"的客户端]执行了" +
                        "[虚拟机"+virtualHost.getVirtualHostName()+"]的[exchangeDeclare]方法");
                ExchangeDeclareArguments arguments1 = (ExchangeDeclareArguments) basicArguments;
                ok = virtualHost.exchangeDeclare(arguments1.getExchangeName(), arguments1.getExchangeType()
                        ,arguments1.isDurable(),arguments1.isAutoDelete(),arguments1.getArguments());
                break;
            //删除交换机
            case 0x4:
                log.info("[process] [IP为"+clientSocket.getInetAddress().toString()+"的客户端]执行了" +
                        "[虚拟机"+virtualHost.getVirtualHostName()+"]的[exchangeDelete]方法");
                ExchangeDeleteArguments arguments2 = (ExchangeDeleteArguments) basicArguments;
                ok = virtualHost.exchangeDelete(arguments2.getExchangeName());
                break;
            //新建队列
            case 0x5:
                log.info("[process] [IP为"+clientSocket.getInetAddress().toString()+"的客户端]执行了" +
                        "[虚拟机"+virtualHost.getVirtualHostName()+"]的[queueDeclare]方法");
                QueueDeclareArguments arguments3 = (QueueDeclareArguments) basicArguments;
                ok = virtualHost.queueDeclare(arguments3.getQueueName(),arguments3.isDurable()
                        ,arguments3.isExclusive(),arguments3.isAutoDelete(),arguments3.getArguments());
                break;
            //删除队列
            case 0x6:
                log.info("[process] [IP为"+clientSocket.getInetAddress().toString()+"的客户端]执行了" +
                        "[虚拟机"+virtualHost.getVirtualHostName()+"]的[queueDelete]方法");
                QueueDeleteArguments arguments4 = (QueueDeleteArguments) basicArguments;
                ok = virtualHost.queueDelete(arguments4.getQueueName());
                break;
            //新建绑定
            case 0x7:
                log.info("[process] [IP为"+clientSocket.getInetAddress().toString()+"的客户端]执行了" +
                        "[虚拟机"+virtualHost.getVirtualHostName()+"]的[queueBind]方法");
                QueueBindingArguments arguments5 = (QueueBindingArguments) basicArguments;
                ok = virtualHost.queueBind(arguments5.getExchangeName(),arguments5.getQueueName(),arguments5.getBindingKey());
                break;
            //解除绑定
            case 0x8:
                log.info("[process] [IP为"+clientSocket.getInetAddress().toString()+"的客户端]执行了" +
                        "[虚拟机"+virtualHost.getVirtualHostName()+"]的[queueUnBind]方法");
                QueueUnBindingArguments arguments6 = (QueueUnBindingArguments) basicArguments;
                ok = virtualHost.queueUnBind(arguments6.getExchangeName(),arguments6.getQueueName());
                break;
            //发送消息
            case 0x9:
                log.info("[process] [IP为"+clientSocket.getInetAddress().toString()+"的客户端]执行了" +
                        "[虚拟机"+virtualHost.getVirtualHostName()+"]的[basicPublish]方法");
                BasicPublishArguments arguments7 = (BasicPublishArguments) basicArguments;
                ok = virtualHost.basicPublish(arguments7.getExchangeName(), arguments7.getRoutingKey(),arguments7.getBasicProperties(),arguments7.getData());
                break;
            //订阅消息
            case 0xa:
                log.info("[process] [IP为"+clientSocket.getInetAddress().toString()+"的客户端]执行了" +
                        "[虚拟机"+virtualHost.getVirtualHostName()+"]的[basicConsume]方法");
                //1.转换消息类型
                BasicConsumeArguments arguments8 = (BasicConsumeArguments) basicArguments;
                //2.虚拟机执行指定方法
                ok = virtualHost.basicConsume(arguments8.getConsumerTag(), arguments8.getQueueName(), arguments8.isAutoAck()
                        , new Consumer() {
                            //丁一回调函数，将消息推送给客户端
                            @Override
                            public void handleDelivery(String consumerTag, BasicProperties basicProperties, byte[] data){

                                try {
                                    //1.获取客户端，从会话中
                                    Socket clientSocket = session.get(consumerTag);
                                    //2.检验客户端
                                    if(clientSocket == null || clientSocket.isClosed()){
                                        new MqException("[BrokerServer] [ID为"+clientSocket.getInetAddress().toString()
                                                +"的客户端]执行回调方法失败,因为客户端已经无连接").printStackTrace();
                                    }
                                    //3.新建返回数据，返回给客户端的
                                    SubScribeReturn subScribeReturn = new SubScribeReturn();
                                    subScribeReturn.setChannelId(consumerTag);
                                    subScribeReturn.setRid("");//此次响应无对应的请求,则无对应的rid
                                    subScribeReturn.setRet(true);
                                    subScribeReturn.setConsumerTag(consumerTag);
                                    subScribeReturn.setBasicProperties(basicProperties);
                                    subScribeReturn.setBody(data);
                                    //4.将返回数据打包成响应，返回给客户端
                                    byte[] payload = SerializableUtils.toBytes(subScribeReturn);
                                    Response response = new Response();
                                    response.setType(0xc); //0xc表示给客户端推送消息
                                    response.setLength(payload.length);
                                    response.setPayload(payload);
                                    //5.写入套接字，将“响应”
                                    //  注意！dataOutputStream不能关闭，因为会触发clientSocket的outStream 关闭
                                    DataOutputStream dataOutputStream = new DataOutputStream(clientSocket.getOutputStream());
                                    writeResponse(response,dataOutputStream);
                                }catch (IOException e) {
                                    log.error("[process] 推送[消息:"+basicProperties.getId()+"]给" +
                                            "[消费者:"+consumerTag+"]时序列化出现异常");
                                    e.printStackTrace();
                                }

                            }
                        });
                break;
            //消息应答
            case 0xb:
                log.info("[process] [IP为"+clientSocket.getInetAddress().toString()+"的客户端]执行了" +
                        "[虚拟机"+virtualHost.getVirtualHostName()+"]的[basicAck]方法");
                BasicAckArguments arguments9 = (BasicAckArguments)basicArguments;
                ok = virtualHost.basicAck(arguments9.getQueueName(),arguments9.getMessageId());
                break;
            default:
                new MqException("[BrokerServer] [process] [ID为"+clientSocket.getInetAddress().toString()+"的客户端]" +
                        "的请求["+request.getType()+"]是未知的类型").printStackTrace();
        }
        //4.构造响应返回的基本信息
        BasicReturns basicReturns = new BasicReturns();
        basicReturns.setRid(basicArguments.getRid());
        basicReturns.setChannelId(basicArguments.getChannelId());
        basicReturns.setRet(ok);

        //5.构造响应，返回给客户端的响应
        byte[] payload = SerializableUtils.toBytes(basicReturns);
        Response response = new Response();
        response.setType(request.getType());
        response.setLength(payload.length);
        response.setPayload(payload);
        return response;
    }

}
